package com.atguigu.state;

/*
 * @date 2024/11/22 10:03
 *
 * 网站中一个非常经典的例子，就是实时统计一段时间内的热门url。
 * 例如，需要统计最近10秒钟内最热门的两个url链接，并且每5秒钟更新一次。
 * 我们知道，这可以用一个滑动窗口来实现，而“热门度”一般可以直接用访问量来表示。
 * 于是就需要开滑动窗口收集url的访问数据，按照不同的url进行统计，而后汇总排序并最终输出前两名。
 * 这其实就是著名的“Top N”问题。
 *
 * todo 思路：使用增量聚合 + 全窗口函数 + 状态 + 定时器  理解定时器基础上写第二遍,未测
 *
 * 1.使用增量聚合 + 全窗口函数完成UrlViewCount
 *
 * 2.按照窗口的结束时间进行keyby，将同一时间范围窗口数出的数据搜集到一起
 *
 * 3.下游算子将输入的数据存入到状态中，并注册窗口结束时间的定时器
 *
 * 4.定时器触发，表示来自于同一个窗口的数据都搜集齐了，对状态中的数据进行排序，输出topN
 *
 *
*
* */

import com.atguigu.pojo.Event;
import com.atguigu.pojo.UrlViewCount;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.hadoop.yarn.webapp.hamlet2.Hamlet;
import util.SourceUtil;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;

public class Flink13_TopN_AggregateFunctionAndProcessWindowFunctionAndStateAndTimer {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //todo 1.对接数据源
        SingleOutputStreamOperator<Event> ds = env.fromSource(SourceUtil.getSource(), WatermarkStrategy.noWatermarks(), "dataGenSource")
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                                .withTimestampAssigner(
                                        (event, ts) -> event.getTs()
                                )
                );
        ds.print("INPUT");

        //todo 2.使用增量聚合统计url点击数
        SingleOutputStreamOperator<UrlViewCount> windowDs = ds
                .keyBy(Event::getUrl)
                .window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5)))
                .aggregate(
                        //1.增量聚合求点击次数
                        new AggregateFunction<Event, UrlViewCount, UrlViewCount>() {
                            @Override
                            public UrlViewCount createAccumulator() {
                                return new UrlViewCount(null, null, null, 0L);
                            }

                            @Override
                            public UrlViewCount add(Event event, UrlViewCount acculumtor) {
                                //累加器实现累加
                                acculumtor.setCount(acculumtor.getCount() + 1);
                                return acculumtor;
                            }

                            @Override
                            public UrlViewCount getResult(UrlViewCount acculumtor) {
                                return acculumtor;
                            }

                            @Override
                            public UrlViewCount merge(UrlViewCount urlViewCount, UrlViewCount acc1) {
                                return null;
                            }
                        },

                        //2.全窗口触动定时器
                        new ProcessWindowFunction<UrlViewCount, UrlViewCount, String, TimeWindow>() {
                            @Override
                            public void process(String key, ProcessWindowFunction<UrlViewCount, UrlViewCount, String, TimeWindow>.Context context, Iterable<UrlViewCount> elements, Collector<UrlViewCount> out) throws Exception {

                                //获取窗口信息
                                long windowStart = context.window().getStart();
                                long windowEnd = context.window().getEnd();

                                //补充key，窗口信息
                                UrlViewCount urlViewCount = elements.iterator().next();

                                urlViewCount.setWindowStart(windowStart);
                                urlViewCount.setWindowEnd(windowEnd);
                                urlViewCount.setUrl(key);

                                //输出
                                out.collect(urlViewCount);

                            }

                        }

                );
        windowDs.print("windowDs");

        //todo 3.按照窗口结束时间进行keyby
        SingleOutputStreamOperator<String> processDs = windowDs
                .keyBy(UrlViewCount::getWindowEnd)
                .process(
                        new KeyedProcessFunction<Long, UrlViewCount, String>() {

                            //1.声明状态
                            private ListState<UrlViewCount> listState;//todo 不同窗口计算的不同url统计结果暂时存在list中

                            private ValueState<Boolean> tsState;//todo 用于存储一个布尔值，作为是否已为当前窗口注册了定时器的标志。

                            //2.open初始化状态
                            @Override
                            public void open(Configuration parameters) throws Exception {
                                //通过上下文对象拿到两个中间状态值
                                ListStateDescriptor<UrlViewCount> listStateDesc = new ListStateDescriptor<UrlViewCount>("listStateDesc", Types.POJO(UrlViewCount.class));
                                ListState<UrlViewCount> listState1 = getRuntimeContext().getListState(listStateDesc);

                                ValueStateDescriptor timerStateDesc = new ValueStateDescriptor("timerStateDesc", Types.BOOLEAN);
                                ValueState valueState = getRuntimeContext().getState(timerStateDesc);

                            }

                            @Override
                            public void processElement(UrlViewCount urlViewCount, KeyedProcessFunction<Long, UrlViewCount, String>.Context context, Collector<String> collector) throws Exception {
                                //3.下游算子将输入的数据存到状态中，并注册窗口结束时间的定时器
                                listState.add(urlViewCount);

                                if (tsState.value() == null) {
                                    context.timerService().registerProcessingTimeTimer(context.getCurrentKey());
                                    tsState.update(true);//用于存储一个布尔值，作为是否已为当前窗口注册了定时器的标志。
                                }
                            }

                            @Override
                            public void onTimer(long timestamp, KeyedProcessFunction<Long, UrlViewCount, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                                //定时器触发，表示来自于同一个窗口的数据都收集齐了，对状态中的数据进行排序，输出topN
                                //从状态中取出所有数据
                                Iterable<UrlViewCount> urlViewCounts = listState.get();

                                //存到可排序的List集合中
                                ArrayList<UrlViewCount> urlViewCountArrayList = new ArrayList<>();
                                urlViewCounts.forEach(urlViewCountArrayList::add);

                                //排序
                                urlViewCountArrayList.sort(
                                        ((o1, o2) -> Long.compare(o2.getCount(), o1.getCount())) //倒序排序
                                );

                                //取出topN
                                //创建可变字符创
                                StringBuilder topNBuilder = new StringBuilder("==================Top2===============\n");
                                //遍历urlViewCountArrayList输出topN
                                for (int i = 0; i < Math.min(2, urlViewCountArrayList.size()); i++) {

                                    topNBuilder.append("Top." + (i + 1) + " " + urlViewCountArrayList.get(i) + "\n");
                                }
                                topNBuilder.append("=====================================\n");

                                //输出
                                out.collect(topNBuilder.toString());
                            }


                        }
                );
        processDs.print("topN");


        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

}
